package com.ustcinfo.study.hbase.xiongjiangfeng;

import org.apache.hadoop.hbase.client.Put;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import com.ustcinfo.ishare.bdp.hbase.HBaseOperationDao;

import static org.junit.Assert.*;

public class HbaseTestTest
{
    private static String driver;
    private static String url;
    private static String username;
    private static String password;

    private static ResultSet resultSet;
    private static PreparedStatement preparedStatement;

    static
    {
        driver = "com.mysql.jdbc.Driver";
        url = "jdbc:mysql://localhost:3306/hbase_test";
        username = "root";
        password = "root";
    }

    @Before
    public void init()
    {
        try
        {
            // 初始化mysql-jdbc
            Class.forName(driver);
            Connection connection = DriverManager.getConnection(url, username, password);
            preparedStatement = connection.prepareStatement("SELECT * FROM transfer_task_to");
            resultSet = preparedStatement.executeQuery();
        } catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    @Test
    public void hbaseTest()
    {
        List<Put> list = new ArrayList<>();
        try
        {
            while (resultSet.next())
            {
                // 获取各字段的值
                // F1
                String id = resultSet.getString("id");
                String tId = resultSet.getString("task_id");
                String createTime = resultSet.getString("create_time");
                String startTime = resultSet.getString("start_time");
                String outFilterNumber = resultSet.getString("out_filter_number");
                String fromRunParam = resultSet.getString("from_run_param");

                // F2
                String currentRetryTimes = resultSet.getString("current_retry_times");
                String retryFlag = resultSet.getString("retry_flag");
                String parentTaskId = resultSet.getString("parent_task_id");
                String endTime = resultSet.getString("end_time");
                String killTime = resultSet.getString("kill_time");
                String failureTime = resultSet.getString("failure_time");


                Put put = new Put(id.getBytes());
                /*
                 插入F1数据,为空不插入
                  */
                if (id != null)
                {
                    put.addColumn("F1".getBytes(), "id".getBytes(), id.getBytes());
                }
                if (tId != null)
                {
                    put.addColumn("F1".getBytes(), "tId".getBytes(), tId.getBytes());
                }
                if (createTime != null)
                {
                    put.addColumn("F1".getBytes(), "createTime".getBytes(), createTime.getBytes());
                }
                if (startTime != null)
                {
                    put.addColumn("F1".getBytes(), "startTime".getBytes(), startTime.getBytes());
                }
                if (outFilterNumber != null)
                {
                    put.addColumn("F1".getBytes(), "outFilterNumber".getBytes(), outFilterNumber.getBytes());
                }
                if (fromRunParam != null)
                {
                    put.addColumn("F1".getBytes(), "fromRunParam".getBytes(), fromRunParam.getBytes());
                }

                /*
                插入F2数据
                 */
                if(currentRetryTimes!=null){
                    put.addColumn("F2".getBytes(), "current_retry_times".getBytes(), currentRetryTimes.getBytes());
                }
                if(retryFlag!=null){
                    put.addColumn("F2".getBytes(), "retry_flag".getBytes(), retryFlag.getBytes());
                }
                if(parentTaskId!=null){
                    put.addColumn("F2".getBytes(), "parent_task_id".getBytes(), parentTaskId.getBytes());
                }
                if(endTime!=null){
                    put.addColumn("F2".getBytes(), "end_time".getBytes(), endTime.getBytes());
                }
                if(killTime!=null){
                    put.addColumn("F2".getBytes(), "kill_time".getBytes(), killTime.getBytes());
                }
                if(failureTime!=null){
                    put.addColumn("F2".getBytes(), "failure_time".getBytes(), failureTime.getBytes());
                }

                // 导入Hbase
                list.add(put);
                HBaseOperationDao.putList("DB_TEST",list);
            }

        } catch (Exception e)
        {
            e.printStackTrace();
        }
    }

    @After
    public void closeCon()
    {
        try
        {
            resultSet.close(); // 关闭rs
        } catch (SQLException e)
        {
            e.printStackTrace();
        }
    }
}